package com.gj;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

public class DistributedLock {

    private CuratorFramework client;

    private RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

    /**
     * /lock/product
     */
    private String basePath;

    /**
     * thread-1 -> /lock/product-000000001
     * 存储线程和节点路径的对应关系的
     */
    private final ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();


    public DistributedLock(String path) {
        this.basePath = path;
        this.client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").sessionTimeoutMs(1000 * 60 * 60).retryPolicy(retryPolicy).build();
        //开启连接
        client.start();
    }

    public void lock() throws Exception {
        //创建临时顺序节点
        String currentNodePath = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(basePath);
        map.put(Thread.currentThread().getName(), currentNodePath);
        //split[1] = lock, split[2] = product-000000001
        String[] split = currentNodePath.split("/");
        //例如 product-000000001
        String currentNode = split[2];
        //查 /lock 下的子节点
        List<String> children = client.getChildren().forPath("/" + split[1]);
        //product-00000001,product-0000002 ...
        List<String> nodeList = children.stream().sorted().collect(Collectors.toList());

        int index = -1;
        for (int i = 0; i < nodeList.size(); i++) {
            if (currentNode.equals(nodeList.get(i))) {
                index = i;
                break;
            }
        }

        if (index == -1) {
            throw new IllegalStateException("当前节点未找到");
        }

        if (index == 0) {
            //拿到锁了
            System.out.println("成功获取锁: " + currentNodePath);
        } else {
            String prevNodePath = "/" + split[1] + "/" + nodeList.get(index - 1);
            synchronized (this) {
                //监听上一个节点，进入阻塞状态
                //指定监听的节点路径
                NodeCache nodeCache = new NodeCache(client, prevNodePath);

                //添加监听器
                nodeCache.getListenable().addListener(() -> {
                    ChildData currentData = nodeCache.getCurrentData();
                    if (currentData == null) {
                        System.out.println("节点 " + prevNodePath + " 已被删除！");
                        synchronized (this) {
                            //唤醒等待的线程
                            System.out.println(prevNodePath + "唤醒下一个节点");
                            notify();
                        }
                    }
                });
                //启动监听
                nodeCache.start(true); // true 表示首次启动时尝试获取节点当前数据
                //阻塞等待唤醒
                wait();
            }
            System.out.println(currentNodePath + "被唤醒后拿到锁了");
        }
    }

    public void unlock() {
        try {
            String currentNodePath = map.get(Thread.currentThread().getName());
            Stat stat = client.checkExists().forPath(currentNodePath);
            if (stat != null) {
                client.delete().forPath(currentNodePath);
                System.out.println("释放锁并删除节点: " + currentNodePath);
            }
        } catch (Exception e) {
            throw new RuntimeException("解锁失败", e);
        }
    }
}
